use async_trait::async_trait;
use fluviox_shared::{Result, FluvioXError, models::DataRecord};
use crate::traits::{SourceConnector, ConnectorConfig, ConnectorStatus, ConnectorMetrics};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::path::PathBuf;
use uuid::Uuid;
use chrono::Utc;
use tracing::{info, error, warn};
use tokio::fs::File;
use tokio::io::{AsyncBufReadExt, BufReader};

/// 文件数据源配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileSourceConfig {
    /// 文件路径或目录路径
    pub path: String,
    /// 文件格式
    pub format: FileFormat,
    /// 是否监控文件变化
    pub watch_mode: bool,
    /// 轮询间隔（秒，仅在watch_mode为true时有效）
    pub poll_interval: u64,
    /// 编码格式
    pub encoding: String,
    /// CSV特定配置
    pub csv_config: Option<CsvConfig>,
    /// JSON特定配置
    pub json_config: Option<JsonConfig>,
    /// 文件过滤模式（支持通配符）
    pub file_pattern: Option<String>,
    /// 是否递归扫描子目录
    pub recursive: bool,
}

/// 文件格式
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum FileFormat {
    /// JSON格式（每行一个JSON对象）
    JsonLines,
    /// 标准JSON格式
    Json,
    /// CSV格式
    Csv,
    /// 纯文本格式
    Text,
    /// XML格式
    Xml,
}

/// CSV配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CsvConfig {
    /// 分隔符
    pub delimiter: char,
    /// 是否有标题行
    pub has_header: bool,
    /// 引用字符
    pub quote_char: char,
    /// 转义字符
    pub escape_char: Option<char>,
}

/// JSON配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JsonConfig {
    /// 数据路径（JSONPath）
    pub data_path: Option<String>,
    /// 是否展平嵌套对象
    pub flatten: bool,
}

impl ConnectorConfig for FileSourceConfig {
    fn validate(&self) -> Result<()> {
        if self.path.is_empty() {
            return Err(FluvioXError::Validation("文件路径不能为空".to_string()));
        }
        
        if self.watch_mode && self.poll_interval == 0 {
            return Err(FluvioXError::Validation("监控模式下轮询间隔必须大于0".to_string()));
        }
        
        // 验证编码格式
        if !["utf-8", "gbk", "gb2312", "ascii"].contains(&self.encoding.to_lowercase().as_str()) {
            return Err(FluvioXError::Validation("不支持的编码格式".to_string()));
        }
        
        Ok(())
    }
    
    fn to_json(&self) -> Result<Value> {
        Ok(serde_json::to_value(self)?)
    }
    
    fn from_json(value: Value) -> Result<Self> {
        Ok(serde_json::from_value(value)?)
    }
}

/// 文件数据源连接器
pub struct FileSourceConnector {
    config: Option<FileSourceConfig>,
    status: ConnectorStatus,
    metrics: ConnectorMetrics,
    processed_files: HashMap<PathBuf, u64>, // 文件路径 -> 已处理的行数
    current_files: Vec<PathBuf>,
}

impl FileSourceConnector {
    pub fn new() -> Self {
        Self {
            config: None,
            status: ConnectorStatus::Uninitialized,
            metrics: ConnectorMetrics::default(),
            processed_files: HashMap::new(),
            current_files: Vec::new(),
        }
    }
    
    /// 扫描文件
    async fn scan_files(&self, config: &FileSourceConfig) -> Result<Vec<PathBuf>> {
        let path = PathBuf::from(&config.path);
        let mut files = Vec::new();
        
        if path.is_file() {
            files.push(path);
        } else if path.is_dir() {
            files.extend(self.scan_directory(&path, config).await?);
        } else {
            return Err(FluvioXError::Internal(format!("路径不存在: {}", config.path)));
        }
        
        Ok(files)
    }
    
    /// 扫描目录（非递归版本以避免Send问题）
    async fn scan_directory(&self, dir: &PathBuf, config: &FileSourceConfig) -> Result<Vec<PathBuf>> {
        let mut files = Vec::new();
        let mut entries = tokio::fs::read_dir(dir).await
            .map_err(|e| FluvioXError::Io(e))?;

        while let Some(entry) = entries.next_entry().await
            .map_err(|e| FluvioXError::Io(e))? {
            let path = entry.path();

            if path.is_file() {
                // 检查文件模式匹配
                if let Some(pattern) = &config.file_pattern {
                    if let Some(file_name) = path.file_name().and_then(|n| n.to_str()) {
                        if !self.matches_pattern(file_name, pattern) {
                            continue;
                        }
                    }
                }
                files.push(path);
            } else if path.is_dir() && config.recursive {
                // 简化处理：暂时不支持递归扫描以避免复杂的异步递归
                info!("跳过子目录: {:?} (递归扫描暂未实现)", path);
            }
        }

        Ok(files)
    }
    
    /// 简单的通配符匹配
    fn matches_pattern(&self, filename: &str, pattern: &str) -> bool {
        // 简化实现，只支持 * 通配符
        if pattern == "*" {
            return true;
        }
        
        if pattern.contains('*') {
            let parts: Vec<&str> = pattern.split('*').collect();
            if parts.len() == 2 {
                let prefix = parts[0];
                let suffix = parts[1];
                return filename.starts_with(prefix) && filename.ends_with(suffix);
            }
        }
        
        filename == pattern
    }
    
    /// 读取文件内容
    async fn read_file_records(&mut self, file_path: &PathBuf, config: &FileSourceConfig, batch_size: usize) -> Result<Vec<DataRecord>> {
        let file = File::open(file_path).await
            .map_err(|e| FluvioXError::Io(e))?;
        let reader = BufReader::new(file);
        let mut lines = reader.lines();
        
        let mut records = Vec::new();
        let mut line_number = 0;
        let start_line = self.processed_files.get(file_path).unwrap_or(&0);
        
        // 跳过已处理的行
        while line_number < *start_line {
            if lines.next_line().await
                .map_err(|e| FluvioXError::Io(e))?.is_none() {
                break;
            }
            line_number += 1;
        }
        
        // 读取新行
        let mut processed_count = 0;
        while let Some(line) = lines.next_line().await
            .map_err(|e| FluvioXError::Io(e))? {
            if processed_count >= batch_size {
                break;
            }
            
            line_number += 1;
            
            match self.parse_line(&line, config, line_number) {
                Ok(Some(record)) => {
                    records.push(record);
                    processed_count += 1;
                },
                Ok(None) => {
                    // 跳过空行或无效行
                },
                Err(e) => {
                    warn!("解析行失败 (文件: {:?}, 行: {}): {}", file_path, line_number, e);
                    self.metrics.errors_count += 1;
                }
            }
        }
        
        // 更新已处理的行数
        self.processed_files.insert(file_path.clone(), line_number);
        
        Ok(records)
    }
    
    /// 解析单行数据
    fn parse_line(&self, line: &str, config: &FileSourceConfig, line_number: u64) -> Result<Option<DataRecord>> {
        if line.trim().is_empty() {
            return Ok(None);
        }
        
        let data = match config.format {
            FileFormat::JsonLines => {
                serde_json::from_str::<Value>(line)?
            },
            FileFormat::Json => {
                // 对于标准JSON，整个文件应该是一个JSON对象或数组
                // 这里简化处理，按行解析
                serde_json::from_str::<Value>(line)?
            },
            FileFormat::Text => {
                Value::String(line.to_string())
            },
            FileFormat::Csv => {
                // 简化的CSV解析
                let csv_config = config.csv_config.as_ref()
                    .ok_or_else(|| FluvioXError::Validation("CSV格式需要CSV配置".to_string()))?;
                
                let fields: Vec<&str> = line.split(csv_config.delimiter).collect();
                let mut obj = serde_json::Map::new();
                
                for (i, field) in fields.iter().enumerate() {
                    obj.insert(format!("field_{}", i), Value::String(field.to_string()));
                }
                
                Value::Object(obj)
            },
            FileFormat::Xml => {
                // XML解析需要额外的库，这里简化处理
                Value::String(line.to_string())
            },
        };
        
        let mut metadata = HashMap::new();
        metadata.insert("line_number".to_string(), line_number.to_string());
        metadata.insert("source_file".to_string(), "file".to_string());
        
        Ok(Some(DataRecord {
            id: Uuid::new_v4(),
            topic: "file-source".to_string(),
            partition: 0,
            offset: line_number,
            timestamp: Utc::now(),
            data,
            metadata,
        }))
    }
}

#[async_trait]
impl SourceConnector for FileSourceConnector {
    type Config = FileSourceConfig;
    
    fn name(&self) -> &str {
        "file-source"
    }
    
    fn version(&self) -> &str {
        "1.0.0"
    }
    
    fn description(&self) -> &str {
        "文件数据源连接器，支持JSON、CSV、文本等格式，支持目录监控"
    }
    
    async fn initialize(&mut self, config: Self::Config) -> Result<()> {
        info!("初始化文件数据源连接器");
        
        // 验证配置
        config.validate()?;
        
        // 扫描初始文件
        self.current_files = self.scan_files(&config).await?;
        info!("发现 {} 个文件", self.current_files.len());
        
        self.config = Some(config);
        self.status = ConnectorStatus::Initialized;
        
        info!("文件数据源连接器初始化完成");
        Ok(())
    }
    
    async fn test_connection(&self) -> Result<bool> {
        info!("测试文件连接");
        
        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("连接器未初始化".to_string()))?;
        
        let path = PathBuf::from(&config.path);
        
        if path.exists() {
            if path.is_file() {
                // 尝试打开文件
                match File::open(&path).await {
                    Ok(_) => {
                        info!("文件连接测试成功");
                        Ok(true)
                    },
                    Err(e) => {
                        error!("文件连接测试失败: {}", e);
                        Ok(false)
                    }
                }
            } else if path.is_dir() {
                // 尝试读取目录
                match tokio::fs::read_dir(&path).await {
                    Ok(_) => {
                        info!("目录连接测试成功");
                        Ok(true)
                    },
                    Err(e) => {
                        error!("目录连接测试失败: {}", e);
                        Ok(false)
                    }
                }
            } else {
                warn!("路径既不是文件也不是目录");
                Ok(false)
            }
        } else {
            warn!("路径不存在: {}", config.path);
            Ok(false)
        }
    }
    
    async fn start(&mut self) -> Result<()> {
        info!("启动文件数据源连接器");
        self.status = ConnectorStatus::Running;
        Ok(())
    }
    
    async fn stop(&mut self) -> Result<()> {
        info!("停止文件数据源连接器");
        self.status = ConnectorStatus::Stopped;
        Ok(())
    }
    
    async fn read_records(&mut self, batch_size: usize) -> Result<Vec<DataRecord>> {
        let config = self.config.clone()
            .ok_or_else(|| FluvioXError::Internal("连接器未初始化".to_string()))?;

        // 如果是监控模式，重新扫描文件
        if config.watch_mode {
            self.current_files = self.scan_files(&config).await?;
        }

        let mut all_records = Vec::new();
        let mut remaining_batch_size = batch_size;

        let files_to_process = self.current_files.clone();
        for file_path in &files_to_process {
            if remaining_batch_size == 0 {
                break;
            }

            match self.read_file_records(file_path, &config, remaining_batch_size).await {
                Ok(records) => {
                    remaining_batch_size = remaining_batch_size.saturating_sub(records.len());
                    all_records.extend(records);
                },
                Err(e) => {
                    error!("读取文件失败 {:?}: {}", file_path, e);
                    self.metrics.errors_count += 1;
                }
            }
        }

        // 更新指标
        self.metrics.records_processed += all_records.len() as u64;
        self.metrics.last_activity = Some(Utc::now());

        info!("成功读取 {} 条记录", all_records.len());
        Ok(all_records)
    }
    
    fn status(&self) -> ConnectorStatus {
        self.status.clone()
    }
    
    fn metrics(&self) -> ConnectorMetrics {
        self.metrics.clone()
    }
}
